'use strict';
const Stream = require('stream');
const Readable = Stream.Readable;
const Transform = Stream.Transform;
const Writable = Stream.Writable;
const zlib = require('zlib');
const through = require('through2');
const peek = require('peek-stream');
const async = require('async');
const internals = {
  registry: {},
  settings: {},
  debug: 0
};
function mkname (stream, streamName) {
  if (!mkname.cnt) { mkname.cnt = {}; }
  if (!mkname.cnt[streamName]) { mkname.cnt[streamName] = 0; }
  mkname.cnt[streamName]++;
  stream.arkimeName = streamName + '-' + mkname.cnt[streamName];
}
class ItemTransform extends Transform {
  constructor (options) {
    super({ objectMode: true });
    this._itemTransform = {
      state: 0,
      max: options.maxPeekItems || 1,
      items: []
    };
  }

  _transform (item, encoding, callback) {
    if (!item) {
      return callback();
    }

    const self = this;
    switch (self._itemTransform.state) {
    case 0:
      self._itemTransform.items.push(item);
      if (self._shouldProcess(item)) {
        self._itemTransform.state = 1;
        async.each(self._itemTransform.items, function (eachItem, cb) {
          try {
            self._process(eachItem, function (err, data) {
              if (data) {
                self.push(data);
              }
              return cb();
            });
          } catch (err) {
            cb(err);
            console.log("Couldn't decode", err);
          }
        }, function (err) {
          self._itemTransform.items = [];
          return callback();
        });
        return;
      }
      if (self._itemTransform.items.length === self._itemTransform.max) {
        self._itemTransform.state = 2;
        while ((item = self._itemTransform.items.shift())) {
          self.push(item);
        }
        self._itemTransform.items = [];
      }
      return callback();
    case 1:
      try {
        return self._process(item, callback);
      } catch (err) {
        return callback(err);
      }
    case 2:
      return callback(null, item);
    }
  };

  _flush (callback) {
    if (this._itemTransform.state === 1) {
      return this._finish(callback);
    }

    let item;
    while ((item = this._itemTransform.items.shift())) {
      this.push(item);
    }

    callback();
  };
}
class Pcap2ItemStream extends Readable {
  constructor (options, pcap) {
    super({ objectMode: true });
    mkname(this, 'Pcap2ItemStream');
    this.data = pcap || [];
  }

  _read (size) {
    const data = this.data;
    for (let i = 0; i < data.length; i++) {
      data[i].client = (i % 2);
      this.push(data[i]);
    }
    this.push(null);
  };
}
function createUncompressStream (options, context) {
  return peek({ newline: false, maxBuffer: 3 }, function (data, swap) {
    if (data.length < 3) {
      return swap(null, through());
    }

    if (data[0] === 0x1f && data[1] === 0x8b && data[2] === 8) {
      if (context.headersMap && context.headersMap['content-encoding'] === 'deflate') {
        context.headersMap['content-encoding'] = 'arkime-gzip';
      }
      const s = zlib.createGunzip();
      return swap(null, s);
    }

    /* if (context.headersMap && context.headersMap["content-encoding"] === "deflate") {
      context.headersMap["content-encoding"] = "arkime-deflate";
      return swap(null, zlib.createDeflate());
    } */

    return swap(null, through());
  });
}
class CollectBodyStream extends Writable {
  constructor (collector, item, headerInfo) {
    super();
    mkname(this, 'CollectBodyStream');

    this.collector = collector;
    this.item = item;
    this.headerInfo = headerInfo;

    this.buffers = [];

    this.on('finish', function (err) {
      this.collector.bodyDone(item, Buffer.concat(this.buffers), this.headerInfo);
    });
  }

  _write (chunk, encoding, callback) {
    this.buffers.push(chunk);
    callback(null);
  };
}
const createPipeline = function (options, order, stream, context) {
  const pipes = [stream];
  function link (p) {
    pipes[p].pipe(pipes[p + 1]).on('error', function (err) {
      console.trace('ERROR', order[p], err);
    });
  }
  for (let i = 0; i < order.length; i++) {
    const ClassOrCreate = internals.registry[order[i]];
    if (!ClassOrCreate) {
      console.trace("ERROR - Couldn't find", order[i], 'in decode registry');
      return;
    }
    if (ClassOrCreate.super_ || Object.getOwnPropertyDescriptor(ClassOrCreate, 'prototype')?.writable === false) {
      pipes.push(new ClassOrCreate(options, context));
    } else {
      pipes.push(ClassOrCreate(options, context));
    }
    link(i);
  }
  return pipes;
};
class ItemHTTPStream extends ItemTransform {
  static STATES = {
    start: 1,
    req: 2,
    req_body: 3,
    req_body_chunk: 4,

    res: 5,
    res_body: 6,
    res_body_chunk: 7,

    pass: 8
  };

  options;
  states;
  method;
  code;
  buffers = [];
  contentLength = [];
  transferEncoding = [];
  contentType = ['', ''];
  startPos = [0, 0];
  bodyNum = 0;
  runningStreams = 0;
  itemPos = 0;

  constructor (options) {
    super({ maxPeekItems: 3 });
    mkname(this, 'ItemHTTPStream');
    this.options = options;
    this.states = [ItemHTTPStream.STATES.start, ItemHTTPStream.STATES.start];
  }

  _shouldProcess (item) {
    return (item.data.length >= 4 && item.data.slice(0, 4).toString() === 'HTTP');
  };

  add (item, endPos) {
    this.itemPos++;

    this.startPos[item.client] = endPos;
    this.push({
      client: item.client,
      ts: item.ts,
      data: item.data.slice(this.startPos[item.client], endPos),
      itemPos: this.itemPos
    });
  }

  processText (item, pos) {
    let endPos = pos;
    while (endPos <= item.data.length && item.data[endPos] !== 0x0a) {
      endPos++;
    }
    let line, upper;
    if (endPos > pos && item.data[endPos - 1] === 0x0d) {
      line = item.data.slice(pos, endPos - 1).toString();
    } else {
      line = item.data.slice(pos, endPos).toString();
    }
    switch (this.states[item.client]) {
    case ItemHTTPStream.STATES.start:
      upper = line;
      this.contentLength[item.client] = 0;
      this.transferEncoding[item.client] = '';
      if (upper.startsWith('CONNECT')) {
        this.states[item.client] = ItemHTTPStream.STATES.req;
        this.method = 'CONNECT';
      } else if (upper.startsWith('HTTP')) {
        this.states[item.client] = ItemHTTPStream.STATES.res;
        const parts = upper.split(/ +/);
        this.code = +parts[1];
      } else if (upper.length > 0) {
        this.states[item.client] = ItemHTTPStream.STATES.req;
        this.method = upper.split(' ')[0];
        const parts = upper.split(/ +/);
        this.url = parts[1];
      }
      break;
    case ItemHTTPStream.STATES.req:
      if (line.length === 0) {
        if (this.method.match(/^(GET|HEAD|DELETE|TRACE)$/)) {
          this.states[item.client] = ItemHTTPStream.STATES.start;
        } else if (this.method.match(/^(CONNECT)$/)) {
          this.states[item.client] = ItemHTTPStream.STATES.pass;
        } else if (this.transferEncoding[item.client] === 'CHUNKED') {
          this.states[item.client] = ItemHTTPStream.STATES.req_body_chunk;
        } else {
          this.states[item.client] = ItemHTTPStream.STATES.req_body;
        }
        this.add(item, endPos + 1);
        break;
      }
      upper = line.toUpperCase();
      if (upper.startsWith('CONTENT-LENGTH')) {
        this.contentLength[item.client] = +upper.substring(15);
      } else if (upper.startsWith('CONTENT-TYPE')) {
        this.contentType[item.client] = upper.substring(14);
      } else if (upper.startsWith('TRANSFER-ENCODING')) {
        this.transferEncoding[item.client] = upper.substring(19);
      }
      break;

    case ItemHTTPStream.STATES.req_body_chunk:
      if (line.length === 0) { break; }
      this.contentLength[item.client] = Number.parseInt(line, 16);
      if (isNaN(this.contentLength[item.client])) { throw new Error('Missing Chunk Length'); }
      if (this.contentLength[item.client] === 0) {
        this.msgEnd(item);
      } else {
        this.states[item.client] = ItemHTTPStream.STATES.res_body;
      }
      break;

    case ItemHTTPStream.STATES.res:
      if (line.length === 0) {
        if (this.code / 100 === 1 || this.code === 204 || this.code === 304) {
          this.states[item.client] = ItemHTTPStream.STATES.start;
        } else if (this.method === undefined) {
        } else if (this.method.match(/^(CONNECT)$/)) {
          this.states[item.client] = ItemHTTPStream.STATES.pass;
        } else if (this.transferEncoding[item.client] === 'CHUNKED') {
          this.states[item.client] = ItemHTTPStream.STATES.res_body_chunk;
        } else {
          this.states[item.client] = ItemHTTPStream.STATES.res_body;
        }
        this.add(item, endPos + 1);
        break;
      }
      upper = line.toUpperCase();
      if (upper.startsWith('CONTENT-LENGTH')) {
        this.contentLength[item.client] = +upper.substring(15);
      } else if (upper.startsWith('CONTENT-TYPE')) {
        this.contentType[item.client] = upper.substring(14);
      } else if (upper.startsWith('TRANSFER-ENCODING')) {
        this.transferEncoding[item.client] = upper.substring(19);
      }
      break;

    case ItemHTTPStream.STATES.res_body_chunk:
      if (line.length === 0) { break; }

      this.contentLength[item.client] = Number.parseInt(line, 16);
      if (isNaN(this.contentLength[item.client])) { throw new Error('Missing Chunk Length'); }
      if (this.contentLength[item.client] === 0) {
        this.msgEnd(item);
      } else {
        this.states[item.client] = ItemHTTPStream.STATES.res_body;
      }
      break;
    } /* switch */

    return endPos + 1;
  }

  msgEnd (item) {
    this.states[item.client] = ItemHTTPStream.STATES.start;
    if (this.bufferStream) {
      this.bufferStream.end();
      delete this.bufferStream;
    }
  }

  processBody (item, pos) {
    if (this.contentLength[item.client]) {
      const avail = item.data.length - pos;
      const used = Math.min(avail, this.contentLength[item.client]);
      if (!this.bufferStream) {
        this.runningStreams++;
        this.bufferStream = new Stream.PassThrough();
        mkname(this.bufferStream, 'bufferStream');
        const info = {
          bodyNum: ++this.bodyNum,
          bodyName: this.url?.split(/[/?=]/).pop(),
          bodyType: 'file',
          itemPos: ++this.itemPos
        };

        if (!info.bodyName || info.bodyName.length === 0) {
          info.bodyName = info.bodyType + info.bodyNum;
        }

        if (this.contentType[item.client].match(/^image/i)) {
          info.bodyType = 'image';
        } else if (this.contentType[item.client].match(/^text/i)) {
          info.bodyType = 'text';
        }

        const order = this.options['ITEM-HTTP'] ? this.options['ITEM-HTTP'].order || [] : [];
        const pipes = createPipeline(this.options, order, this.bufferStream, info);

        item.bodyNum = info.bodyNum;
        item.bodyName = info.bodyName;
        const heb = new CollectBodyStream(this, item, info);
        pipes[pipes.length - 1].pipe(heb);
      }

      this.bufferStream.write(item.data.slice(pos, pos + used));

      pos += used;
      this.contentLength[item.client] -= used;
    }
    if (this.contentLength[item.client] === 0) {
      if (this.transferEncoding[item.client] === 'CHUNKED') {
        this.states[item.client] = ItemHTTPStream.STATES.res_body_chunk;
      } else {
        this.msgEnd(item);
      }
    }
    return pos;
  }

  _process (item, callback) {
    this.startPos[item.client] = 0;
    let pos = 0;
    while (pos < item.data.length) {
      const state = this.states[item.client];
      if (state === ItemHTTPStream.STATES.pass) {
        this.push({ client: item.client, ts: item.ts, data: item.data.slice(0, item.data.length), itemPos: ++this.itemPos });
        break;
      } else if (state === ItemHTTPStream.STATES.req_body || state === ItemHTTPStream.STATES.res_body) {
        pos = this.processBody(item, pos);
      } else {
        pos = this.processText(item, pos);
      }
    }

    callback();
  }

  bodyDone (item, data, headerInfo) {
    this.push({ client: item.client, ts: item.ts, data, bodyNum: headerInfo.bodyNum, bodyType: headerInfo.bodyType, bodyName: headerInfo.bodyName, itemPos: headerInfo.itemPos });
    this.runningStreams--;
    if (this.runningStreams === 0 && this.endCb) {
      this.endCb();
    }
  }

  _finish (callback) {
    if (this.runningStreams > 0) {
      // Since we might not get all the packets, end any current buffer stream
      if (this.bufferStream) {
        this.bufferStream.end();
        delete this.bufferStream;
      }
      this.endCb = callback;
    } else {
      setImmediate(callback);
    }
  }
}
function safeStr(str) {
 if (Array.isArray(str)) { return str.map(x => safeStr(x)); }
    return str.replace(/&/g, '&amp;')
      .replace(/</g, '&lt;')
      .replace(/>/g, '&gt;')
      .replace(/"/g, '&quot;')
      .replace(/'/g, '&#39;')
      .replace(/\//g, '&#47;');
}
function createItemSorterStream (options) {
  const stream = through.obj(function (item, encoding, callback) {
    if (item.itemPos === undefined) {
      item.itemPos = this.items.length;
    }
    this.items.push(item);
    callback();
  }, function (callback) {
    this.items.sort(function (a, b) {
      if (a.ts === b.ts) {
        return a.itemPos - b.itemPos;
      }
      return a.ts - b.ts;
    });

    for (let i = 0; i < this.items.length; i++) {
      this.push(this.items[i]);
    }

    callback();
  });
  stream.items = [];
  return stream;
}
exports.register = function (regName, ClassOrCreate, settings) {
  internals.registry[regName] = ClassOrCreate;
  if (settings) {
    internals.settings[regName] = settings;
  }
};
exports.settings = function () {
  return internals.settings;
};
exports.register('BODY-UNCOMPRESS', createUncompressStream);
exports.register('ITEM-HTTP', ItemHTTPStream);
exports.register('ITEM-SORTER', createItemSorterStream);
exports.register('ITEM-NATURAL', through.ctor({ objectMode: true }, function (item, encoding, callback) {
  if (item.html === undefined) {
    item.html = safeStr(item.data.toString()).replace(/\r?\n/g, '<br>');
  }
  callback(null, item);
}));
exports.register('ITEM-CB', through.ctor({ objectMode: true }, function (item, encoding, callback) {
  if (this.items === undefined) {
    this.items = [];
  }
  this.items.push(item);
  callback();
}, function (callback) {
  this.options['ITEM-CB'].cb(null, this.items);
  callback();
}));

const fs = require('fs');
const incoming = JSON.parse(fs.readFileSync('incoming.json', 'utf-8'));
for (let i = 0; i < incoming.length; i++) {
    incoming[i].data=Buffer.from(incoming[i].data)
}
const options = {
  id: "session.id",
  nodeName: "节点",
  order: [],
  'ITEM-HTTP': {
    order: []
  },
  'ITEM-CB': {
  }
};
options.order.push('ITEM-HTTP');
options.order.push('ITEM-SORTER');
options.order.push('ITEM-NATURAL');
options.order.push('ITEM-CB');
options['ITEM-HTTP'].order.push('BODY-UNCOMPRESS');
options['ITEM-CB'].cb = (err, outgoing) => {
  console.log("结果:")
  console.log(JSON.stringify(outgoing))
};
createPipeline(options, options.order, new Pcap2ItemStream(options, incoming));